package com.dingtone.serialization;

import java.io.UnsupportedEncodingException;
import java.lang.reflect.Field;
import java.lang.reflect.ParameterizedType;
import java.util.Map;

import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Deserializer;

import com.alibaba.fastjson.JSON;

public class JsonDeserializer<T> implements Deserializer<T> {
	private String encoding = "UTF8";
	private Class<T> clz;

	public JsonDeserializer(Class<T> tclz){
		clz = tclz;
	}


	public void configure(Map<String, ?> configs, boolean isKey) {
		String propertyName = isKey ? "key.deserializer.encoding" : "value.deserializer.encoding";
		Object encodingValue = configs.get(propertyName);
		if (encodingValue == null)
			encodingValue = configs.get("deserializer.encoding");
		if (encodingValue != null && encodingValue instanceof String)
			encoding = (String) encodingValue;
	}

	public T deserialize(String topic, byte[] data) {
		try {
            if (data == null)
                return null;
            else{
                String strJson = new String(data, encoding);
                return  JSON.parseObject(strJson,clz);
            }
        } catch (UnsupportedEncodingException e) {
            throw new SerializationException("Error when deserializing byte[] to string due to unsupported encoding " + encoding);
        }
	}

	public void close() {
		// TODO Auto-generated method stub

	}

}
